package events

import (
	"context"
	"github.com/IBM/sarama"
	"go.uber.org/zap"
	"jk-time/webook/pkg/logger"
	"jk-time/webook/pkg/saramax"
	"jk-time/webook/search/service"
	"time"
)

// 通用的 sync data event
// 所有的业务方都可以通过这个 event 来同步数据
type SyncDataEvent struct {
	IndexName string
	DocID     string
	Data      string
}

func (SyncDataEvent) Topic() string {
	return "search_tags"
}

type AnyKafkaComsumer struct {
	client  sarama.Client
	l       logger.Logger
	syncSvc service.SyncService
}

func NewAnyKafkaComsumer(client sarama.Client, l logger.Logger, syncSvc service.SyncService) Consumer {
	return &AnyKafkaComsumer{
		client:  client,
		syncSvc: syncSvc,
		l:       l,
	}
}
func (k *AnyKafkaComsumer) Start() error {
	cg, err := sarama.NewConsumerGroupFromClient("search_tag_1", k.client)
	if err != nil {
		return err
	}
	go func() {
		err := cg.Consume(context.Background(),
			[]string{SyncDataEvent{}.Topic()},
			saramax.NewHandler[SyncDataEvent]("search_tag_1", k.Consume))
		if err != nil {
			zap.L().Error("退出了消费循环,异常", zap.Error(err))
		}
	}()
	return err
}

func (k *AnyKafkaComsumer) Consume(msg *sarama.ConsumerMessage, t SyncDataEvent) error {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	return k.syncSvc.InputAny(ctx, t.IndexName, t.DocID, t.Data)
}
